#pragma once

#include "codec.h"
#include "../../logging/Logging.h"
#include "../../net/EventLoop.h"
#include "../../net/TcpServer.h"
#include "../../mutex/MutexLock.h"
#include "../../thread/ThreadLocalSingleton.h"

#include <map>
#include <set>
#include <stdio.h>
#include <utility>
namespace pubsub
{
    using namespace yhaida;

    typedef std::set<std::string> ConnectionSubscription;

    class Topic
    {
    private:
        std::string _topic;
        std::string _content;
        Timestamp _lastPubTime;
        std::set<TcpConnectionPtr> _audience;

    public:
        Topic(const std::string &topic)
            : _topic(topic)
        {
        }
        void add(const TcpConnectionPtr &conn)
        {
            _audience.insert(conn);
            if (_lastPubTime.valid())
            {
                conn->send(makeMessage());
            }
        }
        void remove(const TcpConnectionPtr &conn)
        {
            _audience.erase(conn);
        }
        void publish(const std::string &content, Timestamp time)
        {
            _content = content;
            _lastPubTime = time;
            std::string message = makeMessage();
            for (auto it = _audience.begin(); it != _audience.end(); ++it)
            {
                (*it)->send(message);
            }
        }

    private:
        std::string makeMessage()
        {
            return "pub " + _topic + "\r\n" + _content + "\r\n";
        }
    };

    class PubSubServerBetter : Utils::nocapyable
    {
    public:
        PubSubServerBetter(EventLoop *loop, const InetAddress &listenAddr);
        ~PubSubServerBetter() = default;

        void start();
        void setThreadNumber(int numthread);

    private:
        void onConnection(const TcpConnectionPtr &conn);
        void onMessage(const TcpConnectionPtr &conn,
                       Buffer *buff,
                       Timestamp receiveTime);
        void timePublish();
        void doSubscribe(const TcpConnectionPtr &conn,
                         const std::string &topic);
        void doUnSubscribe(const TcpConnectionPtr &conn,
                           const std::string &topic);
        void doPublish(const std::string &source,
                       const std::string &topic,
                       const std::string &content,
                       Timestamp time);
        void distributeMessage(const std::string &topic,
                               const std::string &content,
                               Timestamp time);
        Topic &getTopic(const std::string &topic);

        void threadInit(EventLoop *loop);

    public:
        // typedef std::map<std::string, Topic> TOPICS;

        typedef ThreadLocalSingleton<std::map<std::string, Topic>> LocalTopics;

    private:
        EventLoop *_loop;
        TcpServer _server;
        // std::map<std::string, Topic> _topics;

        MutexLock _mutex;
        std::set<EventLoop *> _loops;
    };

}